/****

    activequant - activestocks.eu

    This program is free software; you can redistribute it and/or modify
    it under the terms of the GNU General Public License as published by
    the Free Software Foundation; either version 2 of the License, or
    (at your option) any later version.

    This program is distributed in the hope that it will be useful,
    but WITHOUT ANY WARRANTY; without even the implied warranty of
    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
    GNU General Public License for more details.

    You should have received a copy of the GNU General Public License along
    with this program; if not, write to the Free Software Foundation, Inc.,
    51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.

	
	contact  : contact@activestocks.eu
    homepage : http://www.activestocks.eu

****/
package org.activequant.data.util;

import java.util.HashSet;
import java.util.Iterator;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicReference;

import org.activequant.core.domainmodel.InstrumentSpecification;
import org.activequant.core.domainmodel.SeriesSpecification;
import org.activequant.core.domainmodel.data.MarketDataEntity;
import org.activequant.core.types.TimeFrame;
import org.activequant.core.types.TimeStamp;
import org.activequant.data.retrieval.ISeriesDataIteratorSource;
import org.activequant.data.retrieval.SubscriptionSourceBase;
import org.activequant.util.pattern.events.IEventSink;
import org.apache.log4j.Logger;

/**
 * 
 * a replay base class ... probably created by mike. <br/>
 * An abstract class. ReplayServiceBase&lt;T extends MarketDataEntity&lt;T&gt;&gt; extends SubscriptionSourceBase&lt;T&gt; implements IEventSink&lt;T&gt;. 
 * Defines the DispatchingSubscription class. Holds the following associated variables:
 * <ul>
 * <li>events(Map&lt;SeriesSpecification,IEventSink&lt;T&gt;&gt;)</li>
 * <li>currentTimeStamp(AtomicReference&lt;TimeStamp&gt;)</li>
 * <li>startTimeStamp(TimeStamp)</li>
 * <li>endTimeStamp(TimeStamp)</li>
 * <li>seriesSource(ISeriesDataIteratorSource&lt;T&gt;)</li>
 * <li>activeSeries(Set&lt;SeriesSpecification&gt;)</li>
 * </ul>
 * It also holds the following inherited variables:
 * <ul>
 * <li>activeBackends(Map&lt;Topic, Subscription&gt;)</li>
 * <li>activeSubscriptions(Queue&lt;SubscriptionDelegate&gt;)</li>
 * </ul>
 * <b>History:</b><br>
 *  - [??.??.2008] Created (Mike Kroutikov)<br>
 *
 *  @author Mike Kroutikov
 */
public abstract class ReplayServiceBase<T extends MarketDataEntity<T>> extends SubscriptionSourceBase<T> implements IEventSink<T> {

	private final Logger log = Logger.getLogger(getClass()); 
	/**
	 * An empty ReplayServiceBase&lt;T extends MarketDataEntity&lt;T&gt;&gt; (extends SubscriptionSourceBase&lt;T&gt; implements IEventSink&lt;T&gt;) constructor
	 */
	public ReplayServiceBase() {
		super();
	}
	/**
	 * an abstract method
	 * @param entity
	 * @return
	 */
	protected abstract SeriesSpecification inferSeriesSpecification(T entity); 
	/**
	 * private final Map&lt;SeriesSpecification,IEventSink&lt;T&gt;&gt; events = new ConcurrentHashMap&lt;SeriesSpecification,IEventSink&lt;T&gt;&gt;();
	 */
	private final Map<SeriesSpecification,IEventSink<T>> events = new ConcurrentHashMap<SeriesSpecification,IEventSink<T>>();
	/**
	 * <strong>1.</strong> Maps the given specs(SeriesSpecification)-sink(IEventSink&lt;T&gt;) pair into the associated events(Map&lt;SeriesSpecification,IEventSink&lt;T&gt;&gt;)<br/>
	 * <strong>2.</strong> Creates a copy specs(SeriesSpecification) of the given specs(SeriesSpecification), except it sets its start(TimeStamp) and end(TimeStamp) with the value of the
	 * associated currentTimeStamp(AtomicReference&lt;TimeStamp&gt;) and endTimeStamp(TimeStamp)<br/>
	 * <strong>3.</strong> Adds a copy of the given specs(SeriesSpecification) into the associated activeSeries(Set&lt;SeriesSpecification&gt;). If that SeriesSpecification element already 
	 * existed there, it returns without doing anything else. Otherwise it gets an iterable(Iterable&lt;T&gt;) for the copy specs(SeriesSpecification) using the 
	 * associated seriesSource(ISeriesDataIteratorSource&lt;T&gt;). Then it invokes the abstract <code>processFetchedIterator(iterable.iterator())</code>
	 * @param specs
	 * @param sink
	 */
	private void addEventSink(SeriesSpecification specs, IEventSink<T> sink) {
		events.put(specs, sink);
		fetchSeries(specs);
	}
	/**
	 * removes the mapping for the given specs(SeriesSpecification) from the associated events(Map&lt;SeriesSpecification,IEventSink&lt;T&gt;&gt;)
	 * @param specs
	 */
	private void removeEventSink(SeriesSpecification specs) {
		events.remove(specs);
	}
	/**
	 * A nested class. Extends Subscription implements IEventSink&lt;T&gt;. Holds the following associated variable:
	 * <ul>
	 * <li>specs(SeriesSpecification)</li>
	 * </ul>
	 * It also holds the following inherited variables:
	 * <ul>
	 * <li>isActive(AtomicBoolean)</li>
	 * <li>refCount(AtomicInteger)</li>
	 * <li>instrument(InstrumentSpecification)</li>
	 * <li>event(Event&lt;T&gt;)</li>
	 * </ul>
	 * @author Dimitar
	 *
	 */
	class DispatchingSubscription extends Subscription implements IEventSink<T> {
		/**
		 * private final SeriesSpecification specs;
		 */
		private final SeriesSpecification specs;
		/**
		 * Constructs a DispatchingSubscription (extends Subscription implements IEventSink&lt;T&gt;) using the given ss(SeriesSpecification) to set its
		 * associated specs(SeriesSpecification)
		 * @param ss
		 */
		public DispatchingSubscription(SeriesSpecification ss) {
			specs = ss;
		}
		/**
		 * <strong>1.</strong> Maps this DispatchingSubscription instance into the associated events(Map&lt;SeriesSpecification,IEventSink&lt;T&gt;&gt;) using the associated specs(SeriesSpecification) as key<br/>
		 * <strong>2.</strong> Creates a copy specs(SeriesSpecification) of the associated specs(SeriesSpecification), except it sets its start(TimeStamp) and end(TimeStamp) with the value of the
		 * associated currentTimeStamp(AtomicReference&lt;TimeStamp&gt;) and endTimeStamp(TimeStamp)<br/>
		 * <strong>3.</strong> Adds a copy of the associated specs(SeriesSpecification) into the associated activeSeries(Set&lt;SeriesSpecification&gt;). If that SeriesSpecification element already 
		 * existed there, it returns without doing anything else. Otherwise it gets an iterable(Iterable&lt;T&gt;) for the copy specs(SeriesSpecification) using the 
		 * associated seriesSource(ISeriesDataIteratorSource&lt;T&gt;). Then it invokes the abstract <code>processFetchedIterator(iterable.iterator())</code>
		 */
		@Override
		protected void handleActivate() { 
			addEventSink(specs, this);
		}
		/**
		 * removes the mapping for the associated specs(SeriesSpecification) from the events(Map&lt;SeriesSpecification,IEventSink&lt;T&gt;&gt;)
		 */
		@Override
		protected void handleCancel() { 
			removeEventSink(specs);
		}
		/**
		 * invokes <code>event.fire(data)</code> on the inherited event(Event&lt;T&gt;) using the given e(T)
		 */
		public void fire(T e) {
			super.fireEvent(e);
		}
	}
	/**
	 * <strong>1.</strong> Creates shortSs(SeriesSpecification) using the given spec(InstrumentSpecification) and timeFrame(TimeFrame) to set its associated instrumentSpecification(InstrumentSpecification) and timeFrame(TimeFrame)<br/>
	 * <strong>2.</strong> Creates and returns a DispatchingSubscription(extends Subscription implements IEventSink&lt;T&gt;) using that shortSs(SeriesSpecification) to set its associated specs(SeriesSpecification)
	 */
	@Override
	protected DispatchingSubscription createSubscription(InstrumentSpecification spec, TimeFrame timeFrame) {
		
		SeriesSpecification shortSs = new SeriesSpecification(spec, timeFrame);
		
		return new DispatchingSubscription(shortSs);
	}
	/**
	 * <strong>1.</strong> Finds the spec(SeriesSpecification) for the given e(T) by invoking the abstract <code>inferSeriesSpecification(e)</code> method.<br/>
	 * <strong>2.</strong> Gets the sink(IEventSink&lt;T&gt;) mapped under that spec(SeriesSpecification) into the associated events(Map&lt;SeriesSpecification,IEventSink&lt;T&gt;&gt;)<br/>
	 * <strong>3.</strong> Fires the given e(T) on that sink(IEventSink&lt;T&gt;) by invoking the <code>sink.fire(e)</code> method.
	 */
	public void fire(T e) {
		SeriesSpecification spec = inferSeriesSpecification(e);
		IEventSink<T> sink = events.get(spec);
		if(sink != null) {
			try {
				sink.fire(e);
			} catch(Exception ex) {
				ex.printStackTrace();
				log.error(ex);
			}
		}
	}
	
	/**
	 * Processes the iterator over the fetched market data entities.
	 * Subclass must implement it and do something with the data. Typically, subclass will add this to the
	 * mergeSort to merge and align with the other data streams.<br/>
	 * An abstract method
	 * @param iterator
	 */
	protected abstract void processFetchedIterator(Iterator<T> iterator);
	/**
	 * private final AtomicReference&lt;TimeStamp&gt; currentTimeStamp = new AtomicReference&lt;TimeStamp&gt;();
	 */
	private final AtomicReference<TimeStamp> currentTimeStamp = new AtomicReference<TimeStamp>();
	/**
	 * private TimeStamp startTimeStamp;
	 */
	private TimeStamp startTimeStamp;
	
	/**
	 * Start of the historical interval.<br/>
	 * returns the associated startTimeStamp(TimeStamp)
	 * @return start.
	 */
	public TimeStamp getStartTimeStamp() {
		return startTimeStamp;
	}
	/**
	 * <strong>1.</strong> Sets the associated startTimeStamp(TimeStamp) with the given val(TimeStamp)<br/>
	 * <strong>2.</strong> Sets the value of the associated currentTimeStamp(AtomicReference&lt;TimeStamp&gt;) with the given val(TimeStamp)
	 * @param val
	 */
	public void setStartTimeStamp(TimeStamp val) {
		startTimeStamp = val;
		currentTimeStamp.set(startTimeStamp);
	}
	/**
	 * private TimeStamp endTimeStamp;
	 */
	private TimeStamp endTimeStamp;
	
	/**
	 * End of the historical interval.<br/>
	 * returns the associated endTimeStamp(TimeStamp)
	 * @return end.
	 */
	public TimeStamp getEndTimeStamp() {
		return endTimeStamp;
	}
	/**
	 * sets the associated endTimeStamp(TimeStamp) with the given val(TimeStamp)
	 * @param val
	 */
	public void setEndTimeStamp(TimeStamp val) {
		endTimeStamp = val;
	}
	/**
	 * private ISeriesDataIteratorSource&lt;T&gt; seriesSource;
	 */
	private ISeriesDataIteratorSource<T> seriesSource;
	
	/**
	 * Historic data source for quotes.<br/>
	 * returns the associated seriesSource(ISeriesDataIteratorSource&lt;T&gt;)
	 * @return data source.
	 */
	public ISeriesDataIteratorSource<T> getSeriesDataSource() {
		return seriesSource;
	}
	/**
	 * Sets historic data source for quotes.<br/>
	 * sets the associated seriesSource(ISeriesDataIteratorSource&lt;T&gt;) with the given val(ISeriesDataIteratorSource&lt;T&gt;)
	 * @param val data source.
	 */
	public void setSeriesDataSource(ISeriesDataIteratorSource<T> val) {
		seriesSource = val;
	}
	/**
	 * private final Set&lt;SeriesSpecification&gt; activeSeries = new HashSet&lt;SeriesSpecification&gt;();
	 */
	private final Set<SeriesSpecification> activeSeries = new HashSet<SeriesSpecification>();
	/**
	 * <strong>1.</strong> Creates a copy specs(SeriesSpecification) of the given ss(SeriesSpecification), except it sets its start(TimeStamp) and end(TimeStamp) with the value of the
	 * associated currentTimeStamp(AtomicReference&lt;TimeStamp&gt;) and endTimeStamp(TimeStamp)<br/>
	 * <strong>2.</strong> Adds a copy of the given ss(SeriesSpecification) into the associated activeSeries(Set&lt;SeriesSpecification&gt;). If that SeriesSpecification element already 
	 * existed there, it returns without doing anything else. Otherwise it gets an iterable(Iterable&lt;T&gt;) for the copy specs(SeriesSpecification) using the 
	 * associated seriesSource(ISeriesDataIteratorSource&lt;T&gt;). Then it invokes the abstract <code>processFetchedIterator(iterable.iterator())</code>
	 * @param ss
	 */
	private void fetchSeries(SeriesSpecification ss) {
		if(startTimeStamp == null || endTimeStamp == null) {
			throw new IllegalStateException("backtester must have start and end dates set");
		}
		try {
			SeriesSpecification specs = new SeriesSpecification(ss.getInstrumentSpecification(), ss.getTimeFrame());
			if(activeSeries.add(new SeriesSpecification(specs)) == false) {
				log.info("requested stream is already active: " + specs);
				return;
			}
			specs.setStartTimeStamp(currentTimeStamp.get());
			specs.setEndTimeStamp(endTimeStamp);
			//log.info("fetching data for: " + specs + " from " + seriesSource.getClass());
			Iterable<T> iterable = seriesSource.fetch(specs);
			
			processFetchedIterator(iterable.iterator());
		} catch(Exception ex) {
			ex.printStackTrace();
			log.error(ex);
		}
	}
}
